package com.ffcs.crmd.tsp.server.core;

import java.util.Date;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import org.apache.commons.collections.CollectionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.ffcs.crmd.tsp.api.dto.enu.JobDetailRunStatus;
import com.ffcs.crmd.tsp.api.dto.task.TspJobInstance;
import com.ffcs.crmd.tsp.common.constant.JobDetailDeathResetPolicy;
import com.ffcs.crmd.tsp.common.constant.JobDetailTimeoutPolicy;
import com.ffcs.crmd.tsp.common.constant.LoggerName;
import com.ffcs.crmd.tsp.common.constant.SendStrategy;
import com.ffcs.crmd.tsp.common.enu.JobDetailSendType;
import com.ffcs.crmd.tsp.common.exception.TspException;
import com.ffcs.crmd.tsp.core.thread.ThreadFactoryImpl;
import com.ffcs.crmd.tsp.server.api.Handler;
import com.ffcs.crmd.tsp.server.api.ISelectSendChannel;
import com.ffcs.crmd.tsp.server.application.ApplicationManagerHolder;
import com.ffcs.crmd.tsp.server.config.SchedulerConfig;
import com.ffcs.crmd.tsp.server.jdbc.entity.Job;
import com.ffcs.crmd.tsp.server.jdbc.entity.JobDetail;
import com.ffcs.crmd.tsp.server.jdbc.entity.JobDetailParam;
import com.ffcs.crmd.tsp.server.jdbc.entity.TaskIdentity;
import com.ffcs.crmd.tsp.server.job.info.JobDetailShell;
import com.ffcs.crmd.tsp.server.job.selecter.SelectPollingImpl;
import com.ffcs.crmd.tsp.server.job.selecter.SelectPollingParam;
import com.ffcs.crmd.tsp.server.job.selecter.SelectRebalanceImpl;
import com.ffcs.crmd.tsp.server.job.selecter.SelectRebalanceParam;
import com.ffcs.crmd.tsp.server.manager.JobProcessManager;
import com.ffcs.crmd.tsp.server.manager.TaskRemotingManager;
import com.ffcs.crmd.tsp.server.netty.task.TaskRegisterInfo;
import com.ffcs.crmd.tsp.server.util.PublicMethodHelper;

import io.netty.channel.Channel;

/**
 * 
 * 功能说明:任务调度控制中心
 *
 * @author ZHONGFUHUA
 * 
 * @since 1.0.0-SNAPSHOT
 *
 */
public class SchedulerHandler implements Handler {
    
    /**
     * 日志
     */
    private final static Logger                      LOG                        = LoggerFactory.getLogger(LoggerName.ServerLoggerName);
    
    /**
     * 数据源处理中心
     */
    private DataBaseHandler                          dataBaseHandler;
    
    /**
     * 通信处理中心
     */
    private RemotingHandler                          remotingHandler;
    
    /**
     * quartz处理中心
     */
    private QuartzSchedulerHandler                   quartzSchedulerHandler;
    
    /**
     * 通信服务端配置
     */
    private SchedulerConfig                          schedulerConfig;
    
    /**
     * 轮询发送选择实现
     */
    private ISelectSendChannel<SelectPollingParam>   selectPollingSendChannel   = new SelectPollingImpl();
    
    /**
     * 均衡发送选择实现
     */
    private ISelectSendChannel<SelectRebalanceParam> selectRebalanceSendChannel = new SelectRebalanceImpl();
    
    
    /**
     * 发送任务线程
     */
    private ExecutorService                          sendExecutorService        = Executors
        .newSingleThreadExecutor(new ThreadFactoryImpl("TspServerSendJobThread-"));
    
    /**
     * 定时服务
     */
    private ScheduledExecutorService                 scheduledExecutorService   = Executors.newScheduledThreadPool(5,
        new ThreadFactoryImpl("TspServerScheduledThread-"));
    
    /**
     * 执行端通信管理者
     */
    private TaskRemotingManager                      taskRemotingManager        = ApplicationManagerHolder.getBean("taskRemotingManager",
        TaskRemotingManager.class);
    /**
     * 任务端管理器
     */
    private JobProcessManager                        jobProcessManager          = ApplicationManagerHolder.getBean("jobProcessManager",
        JobProcessManager.class);
    
    public SchedulerHandler(SchedulerConfig schedulerConfig) {
        this.schedulerConfig = schedulerConfig;
    }
    
    /**
     * 
     * 功能说明:
     *
     */
    private void loadAllJobDetail() {
        LOG.error("SERVER-SCHEDULER: loadding all job detail list begin.........");
        try {
            Job jobParam = new Job();
            jobParam.setClusterName(QuartzSchedulerHandler.getClusterName());
            List<Job> jobList = dataBaseHandler.getServiceSupport().queryJob(jobParam);
            for(Job job : jobList){
                JobDetail jobDetailParam = new JobDetail();
                jobDetailParam.setJobId(job.getId());
                List<JobDetail> jobDetailList = dataBaseHandler.getServiceSupport().queryJobDetail(jobDetailParam);
                jobProcessManager.put(job.getName());
                for (JobDetail jobDetail : jobDetailList) {
                    JobDetailParam jobDetailParamParam = new JobDetailParam();
                    jobDetailParamParam.setJobDetailId(jobDetail.getId());
                    List<JobDetailParam> jobDetailParamList = dataBaseHandler.getServiceSupport().queryJobDetailParam(jobDetailParamParam);
                    jobDetail.setParams(jobDetailParamList);
                    quartzSchedulerHandler.addJobDetail(jobDetail, job);
                    LOG.debug("SERVER-SCHEDULER: load a job detail[{}.{}] with[{}].", jobDetail.getJobName(), jobDetail.getName(), jobDetail.getClusterName());
                }
            }
        } catch (Exception e) {
            LOG.debug("SERVER-SCHEDULER: load a job detail exception:", e);
        }
        LOG.debug("SERVER-SCHEDULER: loadding all job detail list end.........");
    }
    
    /**
     * 
     * 功能说明:
     * 
     * @param jobDetailShell
     * @param channel
     */
    private void processTimeoutFinish(JobDetailShell jobDetailShell, Channel channel) {
        TspJobInstance tspJobInstance = jobDetailShell.getContext().getTspJobInstance();
        String jobName = tspJobInstance.getJobName();
        String jobDetailName = tspJobInstance.getName();
        this.remotingHandler.releaseResponse(jobDetailShell.getRemotingOpaque());
        jobProcessManager.completeJobDetail(jobName, channel, jobDetailName, jobDetailShell.getId());
        LOG.warn("SERVER-SCHEDULER: finish a timeout job detail[{}] id=[{}] opaque=[{}] policy=[FINISH]", tspJobInstance.getJobKey(),
            jobDetailShell.getId(), jobDetailShell.getRemotingOpaque());
    }
    
    /**
     * 
     * 功能说明:
     * 
     * @param jobDetailShell
     * @param channel
     */
    private void processTimeoutTransfer(JobDetailShell jobDetailShell, Channel channel) {
        if (jobDetailShell.getStatus() == JobDetailRunStatus.TIMEOUT) {
            boolean result = false;
            TaskRegisterInfo taskRegisterInfo = taskRemotingManager.getTaskRegisterInfo(channel);
            if (PublicMethodHelper.validateOldVersion(taskRegisterInfo)) {
                result = true;
            } else {
                result = remotingHandler.shutdown(jobDetailShell, channel);
            }
            
            if (result) {
                sendJobDetail(jobDetailShell, JobDetailSendType.TRANSFER);
            } else {
                LOG.warn("SERVER-SCHEDULER: transfer a timeout job detail[{}] id=[{}] fail because can't shutdown threads in processing [{}].",
                    jobDetailShell.getJobKey(), jobDetailShell.getId(), channel);
            }
        }
    }
    
    /**
     * 
     * 功能说明:扫描可执行的死任务
     * 
     */
    private void scanJobDetailDeathWillRun() {
        List<String> jobNameList = jobProcessManager.getJobNames();
        for (String jobName : jobNameList) {
            long resetJobDetailDeathDelayTime = jobProcessManager.getResetDeathDelayTime(jobName);
            List<String> jobDetailNameList = jobProcessManager.getJobDetailNameDeaths(jobName);
            for (String jobDetailName : jobDetailNameList) {
                List<JobDetailShell> jobDetailShellList = jobProcessManager.getJobDetailDeath(jobName, jobDetailName);
                for (JobDetailShell jobDetailShell : jobDetailShellList) {
                    if (resetJobDetailDeathDelayTime == JobDetailDeathResetPolicy.IGNORE) {
                        sendJobDetail(jobDetailShell, JobDetailSendType.DEATH);
                    } else if (resetJobDetailDeathDelayTime == JobDetailDeathResetPolicy.RESET) {
                        Date runLastDate = jobProcessManager.getJobDetailTrackLastDate(jobDetailShell);
                        if (System.currentTimeMillis() - runLastDate.getTime() > resetJobDetailDeathDelayTime) {
                            remotingHandler.releaseResponse(jobDetailShell.getRemotingOpaque());
                            jobProcessManager.removeJobDetailDeath(jobDetailShell);
                            jobProcessManager.removeJobDetailTrack(jobDetailShell);
                            jobProcessManager.putJobDetailRetry(jobDetailShell);
                            LOG.info("SERVER-SCHEDULER: death job detail[{}.{}] id=[{}] over delay time will reset put retry to run.", jobName,
                                jobDetailName, jobDetailShell.getId());
                        }
                    }
                }
            }
        }
    }
    
    /**
     * 
     * 功能说明:扫描超时运行子任务
     *
     */
    private void scanJobDetailTimeout() {
        List<String> jobNameList = jobProcessManager.getJobNames();
        if (CollectionUtils.isNotEmpty(jobNameList)) {
            for (String jobName : jobNameList) {
                int timeoutPolicy = jobProcessManager.getTimeoutPolicy(jobName);
                List<JobDetailShell> jobDetailShellList = jobProcessManager.getJobDetailTimeout(jobName);
                for (JobDetailShell jobDetailShell : jobDetailShellList) {
                    Channel lastChannel = jobProcessManager.getJobDetailRunningLastChannel(jobDetailShell);
                    switch (timeoutPolicy) {
                        case JobDetailTimeoutPolicy.FINISH:
                            processTimeoutFinish(jobDetailShell, lastChannel);
                            break;
                        case JobDetailTimeoutPolicy.TRANSFER:
                            processTimeoutTransfer(jobDetailShell, lastChannel);
                            break;
                        default:
                            break;
                    }
                }
            }
        }
    }
    
    /**
     * 
     * 功能说明:扫描待执行的子任务
     * 
     */
    private void scanJobDetailWillRun() {
        List<String> jobNameList = jobProcessManager.getJobNames();
        if (CollectionUtils.isNotEmpty(jobNameList)) {
            for (String jobName : jobNameList) {
                List<JobDetailShell> jobDetailShellList = jobProcessManager.takeJobDetail(jobName, schedulerConfig.getTakeJobDetailBatch());
                if (CollectionUtils.isNotEmpty(jobDetailShellList)) {
                    for (JobDetailShell jobDetailShell : jobDetailShellList) {
                        sendJobDetail(jobDetailShell, JobDetailSendType.NORMAL);
                    }
                }
            }
        }
    }
    
    /**
     * 
     * 功能说明:根据策略或者需要选择发送的通道
     * 
     * @param jobDetailShell 子任务内核
     * @return 通道
     */
    private Channel selectSendChannel(JobDetailShell jobDetailShell) {
        String jobName = jobDetailShell.getJobName();
        switch (schedulerConfig.getSendJobDetailSelectStrategy()) {
            case SendStrategy.POLLING:
                SelectPollingParam selectPollingParam = new SelectPollingParam();
                List<Channel> pollingChannelList = jobProcessManager.getChannels(jobName);
                if (jobDetailShell.getContext().getRefireCount() <= 0) {
                    List<Channel> pollingRanBeforeChannelList = jobProcessManager.getChannelsRunBefore(jobDetailShell);
                    selectPollingParam.setAlreadyChannelList(pollingRanBeforeChannelList);
                }
                selectPollingParam.setChannelRefJob(pollingChannelList);
                selectPollingParam.setJobName(jobName);
                return selectPollingSendChannel.select(selectPollingParam);
            default:
                SelectRebalanceParam selectRebalanceParam = new SelectRebalanceParam();
                Map<Channel, Integer> channelsRunningJobNumber = jobProcessManager.getChannelsRunningNum(jobName);
                if (jobDetailShell.getContext().getRefireCount() <= 0) {
                    List<Channel> rebalanceRanBeforeChannelList = jobProcessManager.getChannelsRunBefore(jobDetailShell);
                    selectRebalanceParam.setRanBeforeChannelList(rebalanceRanBeforeChannelList);
                }
                selectRebalanceParam.setChannelsRunningJobNumberMap(channelsRunningJobNumber);
                return selectRebalanceSendChannel.select(selectRebalanceParam);
        }
    }
    
    /**
     * 
     * 功能说明:发送任务到执行端
     * 
     * @param jobDetailShell
     * @param sendJobDetailType
     */
    private void sendJobDetail(JobDetailShell jobDetailShell, JobDetailSendType sendJobDetailType) {
        TspJobInstance tspJobInstance = jobDetailShell.getContext().getTspJobInstance();
        Channel channel = selectSendChannel(jobDetailShell);
        if (!PublicMethodHelper.validateChannel(channel)) {
            switch (sendJobDetailType) {
                case NORMAL:
                    TspJobInstance tspJobDetail = jobDetailShell.getContext().getTspJobInstance();
                    LOG.warn("SERVER-SCHEDULER: send a {} job detail[{}] id=[{}] channel={} fail because no channel is available or invalid.",
                        tspJobDetail.getJobType(), tspJobDetail.getJobKey(), jobDetailShell.getId(), channel);
                    jobProcessManager.putJobDetailRetry(jobDetailShell);
                    break;
                case DEATH:
                    LOG.warn(
                        "SERVER-SCHEDULER: send a {} death job detail[{}] id=[{}] channel={} fail because no channel is available or invalid.",
                        tspJobInstance.getJobType(), tspJobInstance.getJobKey(), jobDetailShell.getId(), channel);
                    break;
                case TRANSFER:
                    LOG.warn(
                        "SERVER-SCHEDULER: transfer a timeout job detail[{}] id=[{}] channel={} fail because no channel is available or invalid.",
                        tspJobInstance.getJobKey(), jobDetailShell.getId(), channel);
                    Channel lastChannel = jobProcessManager.getJobDetailRunningLastChannel(jobDetailShell);
                    jobProcessManager.removeJobDetailRunning(lastChannel, jobDetailShell);
                    jobProcessManager.putJobDetailDeath(jobDetailShell);
                    break;
            }
            return;
        }
        
        remotingHandler.send(channel, jobDetailShell, sendJobDetailType);
        
        switch (sendJobDetailType) {
            case NORMAL:
                jobProcessManager.putJobDetailRunning(jobDetailShell, channel);
                LOG.debug("SERVER-SCHEDULER: send a {} job detail[{}] id=[{}] to [{}], opaque=[{}].", tspJobInstance.getJobType(),
                    tspJobInstance.getJobKey(), jobDetailShell.getId(), channel, jobDetailShell.getRemotingOpaque());
                break;
            case DEATH:
                jobProcessManager.removeJobDetailDeath(jobDetailShell);
                jobProcessManager.putJobDetailRunning(jobDetailShell, channel);
                LOG.warn("SERVER-SCHEDULER: send a {} death job detail[{}] id=[{}] to [{}], opaque=[{}].", tspJobInstance.getJobType(),
                    tspJobInstance.getJobKey(), jobDetailShell.getId(), channel, jobDetailShell.getRemotingOpaque());
                break;
            case TRANSFER:
                Channel lastChannel = jobProcessManager.getJobDetailRunningLastChannel(jobDetailShell);
                jobProcessManager.removeJobDetailRunning(lastChannel, jobDetailShell);
                jobProcessManager.putJobDetailRunning(jobDetailShell, channel);
                LOG.warn("SERVER-SCHEDULER: transfer a {} timeout job detail[{}] id=[{}] to [{}], opaque=[{}].", tspJobInstance.getJobType(),
                    tspJobInstance.getJobKey(), jobDetailShell.getId(), channel, jobDetailShell.getRemotingOpaque());
                break;
        }
    }
    
    @Override
    public void shutdown() throws TspException {
        scheduledExecutorService.shutdown();
        sendExecutorService.shutdown();
        LOG.info("SERVER-SCHEDULER: Scheduler handler shutdown ok.");
    }
    
    @Override
    public void start() throws TspException {
        startExecutedService();
        loadAllJobDetail();
        LOG.info("SERVER-SCHEDULER: Scheduler handler start ok.");
    }
    
    /**
     * 
     * 功能说明:启动线程池服务
     *
     */
    private void startExecutedService() {
        sendExecutorService.submit(new Runnable() {
            @Override
            public void run() {
                while (true) {
                    try {
                        SchedulerHandler.this.scanJobDetailWillRun();
                        Thread.sleep(schedulerConfig.getTimerScanJobDetail());
                    } catch (Throwable e) {
                        LOG.error("SERVER-SCHEDULER: scan wait running jobDetail exception:", e);
                    }
                }
            }
        });
        
        /**
         * ==============================  定时扫描任务    ==============================
         */
        scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                try {
                    SchedulerHandler.this.scanJobDetailTimeout();
                } catch (Throwable e) {
                    LOG.error("SERVER-SCHEDULER: scan job detail timeout exception:", e);
                }
            }
        }, 1000, schedulerConfig.getTimerScanJobDetailTimeout(), TimeUnit.MILLISECONDS);
        
        scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                try {
                    SchedulerHandler.this.scanJobDetailDeathWillRun();
                } catch (Throwable e) {
                    LOG.error("SERVER-SCHEDULER: scan job detail death exception:", e);
                }
            }
        }, 1000, schedulerConfig.getTimerScanJobDetailDeath(), TimeUnit.MILLISECONDS);
        
        scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                try {
                    SchedulerHandler.this.updateJobConfig();
                } catch (Throwable e) {
                    LOG.error("SERVER-SCHEDULER: update job config exception:", e);
                }
            }
        }, 5000, schedulerConfig.getTimerUpdateJobConfig(), TimeUnit.MILLISECONDS);
        
        scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                try {
                    SchedulerHandler.this.updateTaskConfig();
                } catch (Throwable e) {
                    LOG.error("SERVER-SCHEDULER: update task config exception:", e);
                }
            }
        }, 5000, schedulerConfig.getTimerUpdateTaskConfig(), TimeUnit.MILLISECONDS);
        
        scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                try {
                    SchedulerHandler.this.jobProcessManager.flushJobDetailRetry();
                } catch (Throwable e) {
                    LOG.error("SERVER-SCHEDULER: flush job detail retry exception:", e);
                }
            }
        }, 5000, schedulerConfig.getTimerFlushJobDetailRetry(), TimeUnit.MILLISECONDS);
    }
    
    private void updateJobConfig() {
        List<String> jobList = jobProcessManager.getJobNames();
        for (String jobName : jobList) {
            try {
                Job jobParam = new Job();
                jobParam.setClusterName(QuartzSchedulerHandler.getClusterName());
                jobParam.setName(jobName);
                Job job = dataBaseHandler.getServiceSupport().getJob(jobParam);
                if (job == null) {
                    LOG.error("SERVER-SCHEDULER: update job[{}] clusterName[{}] config error.", jobName,
                        QuartzSchedulerHandler.getClusterName());
                    continue;
                }
                
                jobProcessManager.updateBufferLimitSize(job.getName(), job.getBufferSizeLimit());
                jobProcessManager.updateJobExecuteTimeout(job.getName(), job.getExecuteTimeout());
                jobProcessManager.updateJobTimeoutPolicy(job.getName(), job.getTimeoutPolicy());
                jobProcessManager.updateResetDeathDelayTime(job.getName(), job.getResetDeathDelayTime());
            } catch (Exception e) {
                LOG.error("SERVER-SCHEDULER: update job[{}] config exception:", jobName, e);
            }
        }
    }
    
    /**
     * 
     * 功能说明:
     *
     */
    private void updateTaskConfig() {
        Iterator<Entry<Channel, TaskRegisterInfo>> channelIt = taskRemotingManager.getChannelTable().entrySet().iterator();
        while (channelIt.hasNext()) {
            Entry<Channel, TaskRegisterInfo> channelEntry = channelIt.next();
            Channel channel = channelEntry.getKey();
            TaskRegisterInfo taskRegisterInfo = channelEntry.getValue();
            
            TaskIdentity taskIdentityParam = new TaskIdentity();
            taskIdentityParam.setIdentity(taskRegisterInfo.getIdentity());
            taskIdentityParam.setClusterName(QuartzSchedulerHandler.getClusterName());
            
            try {
                TaskIdentity taskIdentity = dataBaseHandler.getServiceSupport().getTaskIdentity(taskIdentityParam);
                remotingHandler.updateThreadPool(taskRegisterInfo, taskIdentity, channel);
            } catch (TspException e) {
                LOG.error("SERVER-SCHEDULER: update task config exception:", e);
            }
        }
    }
    
    /**
     * 功能说明: 设置任务管理控制中心
     *
     * @param quartzSchedulerHandler 任务管理控制中心 
     */
    public void setQuartzSchedulerHandler(QuartzSchedulerHandler quartzSchedulerHandler) {
        this.quartzSchedulerHandler = quartzSchedulerHandler;
    }
    
    /**
     * 功能说明: 设置任务端管理器
     *
     * @param jobProcessManager 任务端管理器 
     */
    public void setJobProcessManager(JobProcessManager jobProcessManager) {
        this.jobProcessManager = jobProcessManager;
    }
    
    /**
     * 功能说明: 设置执行端通信管理者
     *
     * @param taskRemotingManager 执行端通信管理者 
     */
    public void setTaskRemotingManager(TaskRemotingManager taskRemotingManager) {
        this.taskRemotingManager = taskRemotingManager;
    }
    
    /**
     * 功能说明: 设置数据源处理中心
     *
     * @param dataBaseHandler 数据源处理中心 
     */
    public void setDataBaseHandler(DataBaseHandler dataBaseHandler) {
        this.dataBaseHandler = dataBaseHandler;
    }
    
    /**
     * 功能说明: 设置通信处理中心
     *
     * @param remotingHandler 通信处理中心 
     */
    public void setRemotingHandler(RemotingHandler remotingHandler) {
        this.remotingHandler = remotingHandler;
    }
    
}
